"""
Case Type   : 函数支持指定并行相关参数
Case Name   : 嵌套游标入参场景--单个嵌套游标
Create At   : 2024/10/12
Owner       : @kyleze
Description :
    1.创建测试表
    2.函数声明为parallel_enable,且为srf函数或者管道函数
    3.调用函数是否支持并行
    4.清理环境
Expect      :
    1.成功
    2.成功
    3.支持并行
    4.成功
History     :
"""
import os
import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant


class ParallelEnable(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.table_name = 't_parallel_test020'
        self.type_name = 'type_parallel_test020'
        self.function_name1 = 'f_parallel_test020_1'
        self.function_name2 = 'f_parallel_test020_2'
        self.function_name3 = 'f_parallel_test020_3'
        self.function_name4 = 'f_parallel_test020_4'

    def test_paralle(self):
        step = '----step1:创建测试表 expect:成功----'
        self.log.info(step)
        create_sql = f"drop table if exists {self.table_name} cascade;" \
                     f"create table {self.table_name} " \
                     f"(employee_id number(20), department_id number, " \
                     f"first_name varchar2(30), last_name varchar2(30), " \
                     f"email varchar2(30), phone_number varchar2(30));" \
                     f"begin" \
                     f"   for i in 1..900000 loop" \
                     f"      insert into {self.table_name} values " \
                     f"(i, 60, 'abc', 'def', '203', '203');" \
                     f"end loop;" \
                     f"commit;" \
                     f"end;" \
                     f"create type {self.type_name} as (" \
                     f"    employee_id numeric(20,0)," \
                     f"    department_id numeric," \
                     f"    first_name character varying(30)," \
                     f"    last_name character varying(30)," \
                     f"    email character varying(30)," \
                     f"    phone_number character varying(30));"
        create_res = self.primary_sh.execut_db_sql(create_sql)
        self.log.info(create_res)
        self.assertIn(
            self.constant.CREATE_TABLE_SUCCESS, create_res, '执行失败' + step)
        self.assertIn(self.constant.CREATE_TYPE_SUCCESS_MSG,
                      create_res, '执行失败' + step)
        ana_res = self.primary_sh.execut_db_sql(f"analyze {self.table_name};")
        self.log.info(ana_res)
        self.assertIn(
            self.constant.ANALYZE_SUCCESS_MSG, ana_res, '执行失败' + step)

        step = '----step2:函数声明为parallel_enable,且为SRF函数或者管道函数' \
               'expect:成功----'
        self.log.info(step)
        fun_sql = f"drop function if exists {self.function_name1};" \
                  f"drop function if exists {self.function_name2};" \
                  f"drop function if exists {self.function_name3};" \
                  f"drop function if exists {self.function_name4};" \
                  f"create or replace function {self.function_name1}" \
                  f"(p sys_refcursor) return setof {self.type_name} " \
                  f"parallel_enable (partition p by hash(employee_id)) is" \
                  f"    out_rec {self.type_name} := {self.type_name}" \
                  f"(null, null, null, null, null, null);" \
                  f"begin" \
                  f"    loop" \
                  f"        fetch p into out_rec.employee_id, " \
                  f"out_rec.department_id, out_rec.first_name, " \
                  f"out_rec.last_name, out_rec.email, out_rec.phone_number;" \
                  f"        exit when p%notfound;" \
                  f"        return next out_rec;" \
                  f"    end loop;" \
                  f"    return;" \
                  f"end {self.function_name1};" \
                  f"create or replace function {self.function_name2}" \
                  f"(p sys_refcursor) return setof {self.type_name} " \
                  f"parallel_enable (partition p by any) is" \
                  f"    out_rec {self.type_name} := {self.type_name}" \
                  f"(null, null, null, null, null, null);" \
                  f"begin" \
                  f"    loop" \
                  f"        fetch p into out_rec.employee_id, " \
                  f"out_rec.department_id, out_rec.first_name, " \
                  f"out_rec.last_name, out_rec.email, out_rec.phone_number;" \
                  f"        exit when p%notfound;" \
                  f"        return next out_rec;" \
                  f"    end loop;" \
                  f"    return;" \
                  f"end {self.function_name2};" \
                  f"create type table_{self.type_name} is table of " \
                  f"{self.type_name};" \
                  f"create or replace function {self.function_name3}" \
                  f"(p sys_refcursor) return table_{self.type_name} " \
                  f"pipelined parallel_enable (partition p by any) is" \
                  f"    out_rec {self.type_name} := {self.type_name}" \
                  f"(null, null, null, null, null, null);" \
                  f"begin" \
                  f"    loop" \
                  f"        fetch p into out_rec.employee_id, " \
                  f"out_rec.department_id, out_rec.first_name, " \
                  f"out_rec.last_name, out_rec.email, out_rec.phone_number;" \
                  f"        exit when p%notfound;" \
                  f"        pipe row(out_rec);" \
                  f"    end loop;" \
                  f"end {self.function_name3};" \
                  f"create or replace function {self.function_name4}" \
                  f"(p sys_refcursor) return table_{self.type_name} " \
                  f"pipelined parallel_enable (partition p by " \
                  f"hash(employee_id)) is" \
                  f"    out_rec {self.type_name} := {self.type_name}" \
                  f"(null, null, null, null, null, null);" \
                  f"begin" \
                  f"    loop" \
                  f"        fetch p into out_rec.employee_id, " \
                  f"out_rec.department_id, out_rec.first_name, " \
                  f"out_rec.last_name, out_rec.email, out_rec.phone_number;" \
                  f"        exit when p%notfound;" \
                  f"        pipe row(out_rec);" \
                  f"    end loop;" \
                  f"end {self.function_name4};"
        fun_res = self.primary_sh.execut_db_sql(fun_sql)
        self.log.info(fun_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, fun_res,
                      '执行失败' + step)

        step = '----step3:调用函数是否支持并行 expect:支持并行----'
        self.log.info(step)
        tra_sql = f"set query_dop = 4;" \
                  f"explain (costs off) select *  from {self.table_name};" \
                  f"explain (costs off) select *  from {self.table_name} " \
                  f"where employee_id < 800000;" \
                  f"explain (costs off) select * from {self.function_name1}" \
                  f"(cursor(select *,(cursor(select * from {self.table_name} " \
                  f"where employee_id < 800000))c1 from {self.table_name}));" \
                  f"explain (costs off) select * from {self.function_name2}" \
                  f"(cursor(select *,cursor(select * from {self.table_name})" \
                  f"from {self.table_name} where employee_id < 800000 ));" \
                  f"explain (costs off) select * from {self.function_name3}" \
                  f"(cursor(select *,(cursor(select * from {self.table_name} " \
                  f"where employee_id < 800000))c1 from {self.table_name}));" \
                  f"explain (costs off) select * from {self.function_name4}" \
                  f"(cursor(select *,cursor(select * from {self.table_name})" \
                  f"from {self.table_name} where employee_id < 800000 ));" \
                  f"explain (costs off) select * from {self.function_name2}" \
                  f"(cursor(select employee_id from {self.function_name2} " \
                  f"(cursor(select * from {self.table_name}))));"
        transaction_res = self.primary_sh.execut_db_sql(tra_sql)
        self.log.info(transaction_res)
        self.assertEqual(
            7, transaction_res.count("dop: 1/4"), '执行失败' + step)

    def tearDown(self):
        self.log.info('----清理环境----')
        table_res = \
            self.primary_sh.execut_db_sql(f'drop table {self.table_name};')
        self.log.info(table_res)
        fun_res1 = self.primary_sh.execut_db_sql(
            f'drop function {self.function_name1};')
        self.log.info(fun_res1)
        fun_res2 = self.primary_sh.execut_db_sql(
            f'drop function {self.function_name2};')
        self.log.info(fun_res2)
        fun_res3 = self.primary_sh.execut_db_sql(
            f'drop function {self.function_name3};')
        self.log.info(fun_res3)
        fun_res4 = self.primary_sh.execut_db_sql(
            f'drop function {self.function_name4};')
        self.log.info(fun_res4)
        type_res = self.primary_sh.execut_db_sql(
            f'drop type {self.type_name} cascade;')
        self.log.info(type_res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, table_res, '删表失败')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      fun_res1, '删表失败')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      fun_res2, '删表失败')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      fun_res3, '删表失败')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      fun_res4, '删表失败')
        self.assertIn(self.constant.DROP_TYPE_SUCCESS_MSG, type_res, '删表失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
